package com.atguigu.day05;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<String> result = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
                ctx.collectWithTimestamp(Tuple2.of("a", 1000L), 1000L);
                ctx.collectWithTimestamp(Tuple2.of("a", 2000L), 2000L);
                ctx.collectWithTimestamp(Tuple2.of("a", 3000L), 3000L);
                ctx.collectWithTimestamp(Tuple2.of("a", 4000L), 4000L);
                ctx.emitWatermark(new Watermark(4999L));
                ctx.collectWithTimestamp(Tuple2.of("a", 1000L), 1000L);
            }

            @Override
            public void cancel() {

            }
        })
                .keyBy(x -> true)
                .process(new KeyedProcessFunction<Boolean, Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> value, Context ctx, Collector<String> out) throws Exception {
                        if (value.f1 < ctx.timerService().currentWatermark()) {
                            ctx.output(new OutputTag<Long>("lateElement") {
                            }, value.f1);
                        } else {
                            out.collect("元素正常输出：" + value);
                        }
                    }
                });
        result.print();
        result.getSideOutput(new OutputTag<Long>("lateElement"){}).print("迟到元素的时间");

        env.execute();
    }
}
